

package org.apache.ocean;

import java.util.*;
import java.util.concurrent.locks.*;
import java.util.concurrent.atomic.*;
import java.util.concurrent.*;
import java.util.logging.*;
import org.apache.commons.collections.map.*;

/**
 *
 * @author  jason rutherglen
 */
public class SearchIndexShardPool {
  public static Logger log = Logger.getLogger(SearchIndexShardPool.class.getName());
  private LinkedBlockingQueue<SearchIndexShard> shardQueue;
  private ReferenceMap map = new ReferenceMap(ReferenceMap.HARD, ReferenceMap.WEAK);
  private ReentrantLock lock = new ReentrantLock();
  private SearchIndexShardManager searchDirectoryManager;
  private ScheduledExecutorService timer;
  
  public SearchIndexShardPool(String type, List<SearchIndexShard> initialList, SearchIndexShardManager searchDirectoryManager) {
    this.searchDirectoryManager = searchDirectoryManager;
    shardQueue = new LinkedBlockingQueue<SearchIndexShard>();
    shardQueue.addAll(initialList);
  }
  
  public int size() {
    return map.size();
  }
  
  
  public void addAll(Collection<SearchIndexShard> collection) throws Exception {
    LockUtil.lock(lock);
    try {
      for (SearchIndexShard searchDirectory : collection) {
        boolean added = add(searchDirectory);
        if (!added) return;
      }
    } finally {
      lock.unlock();
    }
  }

  public boolean add(SearchIndexShard shard) throws Exception {
    LockUtil.lock(lock);
    try {
      boolean exists = map.containsKey(shard.getID());
      if (!exists) {
        shardQueue.put(shard);
        map.put(shard.getID(), shard);
        log.log(Level.INFO, "DirectoryPool.add: "+shard+" true");
        return true;
      }
      log.log(Level.INFO, "DirectoryPool.add: "+shard+" false");
      return false;
    } finally {
      lock.unlock();
    }
  }
  
  public boolean remove(String key) {
    log.log(Level.INFO, "DirectoryPool.remove: "+key);
    LockUtil.lock(lock);
    try {
      Iterator<SearchIndexShard> iterator = shardQueue.iterator();
      while (iterator.hasNext()) {
        SearchIndexShard searchDirectory = iterator.next();
        if (searchDirectory.getID().equals(key)) {
          iterator.remove();
          map.remove(searchDirectory.getID());
          return true;
        }
      }
      return false;
    } finally {
      lock.unlock();
    }
  }
  /**
  private void createNewSearchDirectories(String type, int numToCreate) throws Exception {
    log.log(Level.INFO, "createNewSearchDirectories type: "+type+" numToCreate: "+numToCreate+" list.size(): "+shardQueue.size());
    for (int x=0; x < numToCreate; x++) {
      if (shardQueue.size() >= maxShards) return;
      SearchIndexShard newSearchDirectory = searchDirectoryManager.createSearchDirectory(type);
      searchDirectoryManager.addSearchDirectoryNoLock(newSearchDirectory);
      add(newSearchDirectory);
    }
  }
  **/
  private boolean isWriteable(SearchIndexShard shard) {
    if (shard.isClosed() || !shard.typeEquals(ClusterConstants.READ_WRITE)) {
      return false;
    } else {
      return true;
    }
  }
  /**
  private void verifyPoolSize() {
    int size = shardQueue.size();
    log.log(Level.INFO, "getNext size: "+size);
    if (max > size) {
      int numToCreate = max - size;
      try {
        createNewSearchDirectories(type, numToCreate);
      } catch (Throwable ex) {
        log.log(Level.SEVERE, "", ex);
      }
    }
  }
  **/
  public SearchIndexShard getNext() throws Exception {
    LockUtil.lock(lock);
    try {
      SearchIndexShard searchIndexShard = shardQueue.poll(10, TimeUnit.MILLISECONDS);
      if (searchIndexShard == null) {
        throw new ClusterException("none available shardQueue.size: "+shardQueue.size());
      }
      if (!isWriteable(searchIndexShard)) {
        map.remove(searchIndexShard.getID());
        return getNext();
      } else {
        shardQueue.put(searchIndexShard);
      }
      return searchIndexShard;
    } finally {
      lock.unlock();
    }
  }
}
